package com.rocketmq.example;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class E02_AsyncMessage {

    /**
     * 异步消息，监听返回结果
     * 不阻塞当前线程
     */
    public static void main(String[] args) {
        DefaultMQProducer producer = new DefaultMQProducer(ExampleMQConfig.TOPIC.getName());
        producer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        try {
            for (int i = 0; i < 10; i++) {
                Message message = new Message(ExampleMQConfig.TOPIC.getName(), ExampleMQConfig.TAG.getName(),
                        ("message:" + 1).getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(message, new SendCallback() {

                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.err.println(sendResult);
                    }

                    @Override
                    public void onException(Throwable e) {
                        e.printStackTrace();
                        System.err.println("发送失败");
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        try {
            // 异步消息会直接往下走，直接关闭后就无法链接，采取延时关闭策略
            // producer.shutdown();
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            producer.shutdown();
        }

    }

}
